package drds.server.route.parser.core.impl;

import java.sql.SQLNonTransientException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import drds.server.Server;
import drds.server.cache.LayeredCache;
import drds.server.config.ErrorCode;
import drds.server.config.model.Schema;
import drds.server.config.model.Table;
import drds.server.execute_engine.data_handling.AggregateFunctionType;
import drds.server.execute_engine.data_handling.GroupByFilter;
import drds.server.execute_engine.data_handling.OrderByColum;
import drds.server.execute_engine.data_handling.Value;
import drds.server.route.Route;
import drds.server.route.RouteNode;
import drds.server.route.parser.core.ShardInfoVisitor;
import drds.server.route.parser.core.TableColumnValueSetMap;
import drds.server.route.util.RouteUtil;
import drds.server.sqlparser.sqlobject.expression.AggregateFunction;
import drds.server.sqlparser.sqlobject.expression.AllColumn;
import drds.server.sqlparser.sqlobject.expression.Expression;
import drds.server.sqlparser.sqlobject.expression.binary.BinaryExpression;
import drds.server.sqlparser.sqlobject.expression.binary.BinaryOperator;
import drds.server.sqlparser.sqlobject.expression.data_type.Text;
import drds.server.sqlparser.sqlobject.expression.data_type.number.Int;
import drds.server.sqlparser.sqlobject.expression.data_type.number.Numeric;
import drds.server.sqlparser.sqlobject.expression.name.IName;
import drds.server.sqlparser.sqlobject.expression.name.Identifier;
import drds.server.sqlparser.sqlobject.expression.name.TableAliasName$ColumnName;
import drds.server.sqlparser.sqlobject.statement.Statement;
import drds.server.sqlparser.sqlobject.statement.query.GroupBy;
import drds.server.sqlparser.sqlobject.statement.query.IQuery;
import drds.server.sqlparser.sqlobject.statement.query.Limit;
import drds.server.sqlparser.sqlobject.statement.query.Query;
import drds.server.sqlparser.sqlobject.statement.query.SelectItem;
import drds.server.sqlparser.sqlobject.statement.query.SelectStatement;
import drds.server.sqlparser.sqlobject.statement.query.order_by.OrderByItem;
import drds.server.sqlparser.sqlobject.statement.query.order_by.OrderByType;
import drds.server.sqlparser.sqlobject.statement.query.union_query.UnionQuery;
import drds.server.sqlparser.visitor.VisitorUtils;
import drds.server.util.ObjectUtil;

/**
 * <pre>
 * 修改为分表
 * ITable from = query.getFrom();
 * 
 * 				for (Node node : route.getNodes()) {
 * 					Identifier identifier = new Identifier();
 * 					identifier.setParent(from);
 * 					identifier.setName(node.getSubTableName());
 * 					ExpressionTable expressionTable = new ExpressionTable(identifier);
 * 					query.setFrom(expressionTable);//3个表
 * 					node.setNodeSql(statement.toString());
 * 				}
 * </pre>
 * 
 * 
 */
public class SelectParser extends DefaultParser {

	protected boolean isNeedParseOrderBy = true;

	@Override
	public void statementParse(Route route, Schema schema, Statement statement) {
		SelectStatement selectStatement = (SelectStatement) statement;
		IQuery iQuery = selectStatement.getSubQuery().getIQuery();
		if (iQuery instanceof Query) {
			Query query = (Query) selectStatement.getSubQuery().getIQuery();

			parseOrderBy(route, schema, statement, query);
			if ((query.isForUpdate())//
					&& //
					route.isAutoCommit() == false) {
				route.setWriteTrue$ReadFalse(true);
			}

		} else if (iQuery instanceof UnionQuery) {
			UnionQuery unionQuery = (UnionQuery) iQuery;
			IQuery left = (IQuery) unionQuery.getLeft();
			IQuery right = (IQuery) unionQuery.getLeft();
		}
	}

	protected void parseOrderBy(//
			Route route,//
			Schema schema,//
			Statement statement,//
			Query query) {//
		ShardInfoVisitor visitor = new ShardInfoVisitor();
		statement.accept(visitor);
		if (!isNeedParseOrderBy) {
			return;
		}
		Map<String, String> aliasMap = parseGroupBy(route, schema, statement, query);

		if (query.getOrderBy() != null) {
			List<OrderByItem> orderByItemList = query.getOrderBy().getOrderByItemList();
			route.setOrderByColumnMap(//
			buildOrderByColumnMap(orderByItemList, aliasMap));
		}
		isNeedParseOrderBy = false;
	}

	protected Map<String, String> parseGroupBy(//
			Route route,//
			Schema schema,//
			Statement statement,//
			Query query) {//
		Map<String, String> columnNameAliasMap = new HashMap<String, String>();
		Map<String, Integer> mergeColumnAndIndexMapMap = new HashMap<String, Integer>();
		List<String> groupByColumNameList = new ArrayList<String>();
		List<SelectItem> selectItemList = query.getSelectItemList();
		boolean isNeedChangeSql = false;
		int size = selectItemList.size();
		boolean isDistinct = query.getQuantifier() != null;
		for (int i = 0; i < size; i++) {
			SelectItem selectItem = selectItemList.get(i);
			if (selectItem.getExpression() instanceof AggregateFunction) {
				AggregateFunction aggregateFunction = (AggregateFunction) selectItem.getExpression();
				String methodName = aggregateFunction.getMethodName();
				String aggregateColumName = methodName + "(" + aggregateFunction.getArgumentList().get(0) + ")";
				// 转换为Group By
				groupByColumNameList.add(aggregateColumName);

				// 只处理有别名的情况，无别名添加别名，否则某些数据库会得不到正确结果处理
				int mergeType = AggregateFunctionType.getType(methodName);
				/**
				 * 跨分片avg
				 */
				if (AggregateFunctionType.AVG == mergeType && //
						isRoutMultiNode(schema, route)) {
					// 列名
					String columName = methodName + "_" + i;
					/**
					 * sum
					 */
					SelectItem sumSelectItem = new SelectItem();
					String sumColumName = columName + "_sum";
					sumSelectItem.setAlias(sumColumName);
					//
					AggregateFunction sumAggregateFunction = new AggregateFunction("sum");
					ObjectUtil.copyProperties(aggregateFunction, sumAggregateFunction);
					sumAggregateFunction.getArgumentList().addAll(aggregateFunction.getArgumentList());
					sumSelectItem.setExpression(sumAggregateFunction);
					//
					selectItemList.set(i, sumSelectItem);
					mergeColumnAndIndexMapMap.put(sumColumName, AggregateFunctionType.SUM);
					//
					groupByColumNameList.add(sumColumName);
					groupByColumNameList.add(selectItem.getAlias() != null ? selectItem.getAlias() : ""); // Added
					/**
					 * count
					 */
					SelectItem countSelectItem = new SelectItem();
					String countColumName = columName + "_count";
					countSelectItem.setAlias(countColumName);
					//
					AggregateFunction countAggregateFunction = new AggregateFunction("count");
					ObjectUtil.copyProperties(aggregateFunction, countAggregateFunction);
					countAggregateFunction.getArgumentList().addAll(aggregateFunction.getArgumentList());
					countSelectItem.setExpression(countAggregateFunction);
					//
					selectItemList.add(countSelectItem);
					mergeColumnAndIndexMapMap.put(countColumName, AggregateFunctionType.COUNT);
					//
					{
						isNeedChangeSql = true;
						mergeColumnAndIndexMapMap.put(columName, mergeType);
						route.setHasAggregateColumn(true);
					}

				} else if (AggregateFunctionType.UNSUPPORT != mergeType) {
					if (selectItem.getAlias() != null && //
							selectItem.getAlias().length() > 0) {//
						mergeColumnAndIndexMapMap.put(selectItem.getAlias(), mergeType);
					} else { // 如果不加，jdbc方式时取不到正确结果 ;修改添加别名
						selectItem.setAlias(methodName + "_" + i);
						mergeColumnAndIndexMapMap.put(methodName + "_" + i, mergeType);
						isNeedChangeSql = true;
					}
					route.setHasAggregateColumn(true);
					groupByColumNameList.add(selectItem.getAlias());
					groupByColumNameList.add("");
				}
			} else {
				if (!(selectItem.getExpression() instanceof AllColumn)) {
					String columnName = getSelectItem(selectItem);
					String alias = selectItem.getAlias();

					if (alias == null) {
						alias = columnName;
					}
					columnNameAliasMap.put(columnName, alias);
				}
			}

		}
		if (mergeColumnAndIndexMapMap.size() > 0) {
			route.setMergeColumnAndIndexMap(mergeColumnAndIndexMapMap);
		}

		if (isDistinct) {
			/**
			 * 通过优化转换成group by来实现
			 */
			query.setQuantifier(null);
			GroupBy groupBy = new GroupBy();
			for (String columnName : columnNameAliasMap.keySet()) {
				groupBy.addGroupByItem(new Identifier(columnName));
			}
			query.setGroupBy(groupBy);
			isNeedChangeSql = true;
		}

		if (query.getGroupBy() != null) {
			List<Expression> groupByItemList = query.getGroupBy().getGroupByItemList();
			String[] groupByItemColumns = buildGroupByColumns(groupByItemList, columnNameAliasMap);
			//
			route.setGroupByItemColumns(groupByItemColumns);
			route.setGroupByFilter(buildGroupByFilter(query.getGroupBy().getFilter()));
			route.setHasAggregateColumn(true);
			route.setGroupByFilterColumNames(groupByColumNameList.toArray());
		}

		if (isNeedChangeSql) {
			String sql = statement.toString();
			route.updateLimit(sql, 0, -1);
			getParserResult().setSql(sql);
		}
		return columnNameAliasMap;
	}

	private GroupByFilter buildGroupByFilter(Expression filter) {
		if (filter == null) {
			return null;
		}

		BinaryExpression binaryExpression = ((BinaryExpression) filter);
		Expression left = binaryExpression.getLeft();
		BinaryOperator binaryOperator = binaryExpression.getBinaryOperator();
		Expression right = binaryExpression.getRight();

		String leftValue = null;
		if (left instanceof AggregateFunction) {
			leftValue = ((AggregateFunction) left).getMethodName() + "(" + ((AggregateFunction) left).getArgumentList().get(0) + ")";
		} else if (left instanceof Identifier) {
			leftValue = ((Identifier) left).getName();
		}

		String rightValue = null;
		if (right instanceof Numeric) {
			rightValue = right.toString();
		} else if (right instanceof Text) {
			rightValue = right.toString();
		}

		return new GroupByFilter(leftValue, binaryOperator.getName(), rightValue);
	}

	private boolean isRoutMultiNode(Schema schema, Route route) {
		if (route.getRouteNodes() != null && route.getRouteNodes().length > 1) {
			return true;
		}
		LayeredCache cache = (LayeredCache) Server.getInstance().getCacheService().getCachePool("tableid2datanodecache");
		tryRoute(route, schema, cache);
		if (route.getRouteNodes() != null && route.getRouteNodes().length > 1) {
			return true;
		}
		return false;
	}

	private String getSelectItem(SelectItem selectItem) {//
		if ((selectItem.getExpression() instanceof TableAliasName$ColumnName) || //
				(selectItem.getExpression() instanceof Identifier) || //
				selectItem.getExpression() instanceof BinaryExpression) {//
			return selectItem.getExpression().toString();// 字段别名
		} else {
			return selectItem.toString();
		}
	}

	/**
	 * 改写sql：需要加limit的加上
	 */
	@Override
	public void changeSql(Schema schema, Route route, Statement statement, LayeredCache cache) throws SQLNonTransientException {

		tryRoute(route, schema, cache);

		route.copyLimitToNodes();

		SelectStatement selectStatement = (SelectStatement) statement;
		IQuery iQuery = selectStatement.getSubQuery().getIQuery();
		// limit 只能加在最外层
		if (iQuery instanceof Query) {
			Query query = (Query) selectStatement.getSubQuery().getIQuery();
			int limitOffset = 0;
			int limitCount = schema.getDefaultMaxLimit();

			GroupBy groupBy = query.getGroupBy();
			/**
			 * 当在多个节点上面不允许执行GROUP BY HAVING 操作。可以在数据都到了内存才进行聚合处理。
			 */
			// clear group having if this sql execute at multiple nodes
			if (groupBy != null && groupBy.getFilter() != null && isRoutMultiNode(schema, route)) {
				groupBy.setFilter(null);
			}

			Map<String, Map<String, Set<Value>>> tableColumnValueSetMapMap = getTableColumnValueSetMapMap();
			boolean isNeedAddLimit = isNeedAddLimit(route, schema, query, tableColumnValueSetMapMap);
			if (isNeedAddLimit) {
				Limit limit = new Limit();
				limit.setCount(new Int(limitCount));
				query.setLimit(limit);
				route.setLimitSize(limitCount);
				String sql = getSql(route, statement, isNeedAddLimit);
				route.updateLimit(sql, 0, limitCount);

			}
			Limit limit = query.getLimit();
			if (limit != null && !isNeedAddLimit) {
				Int offset = (Int) limit.getOffset();
				Int count = (Int) limit.getCount();
				if (offset != null) {
					limitOffset = offset.getNumber().intValue();
					route.setLimitOffset(limitOffset);
				}
				if (count != null) {
					limitCount = count.getNumber().intValue();
					route.setLimitSize(limitCount);
				}

				if (isNeedChangeLimit(route)) {
					Limit changedLimit = new Limit();
					changedLimit.setCount(new Int(limitOffset + limitCount));

					if (offset != null) {
						if (limitOffset < 0) {
							String msg = "You have an error in your SQL syntax; check the manual that " + "corresponds to your MySQL server version for the right syntax to use near '" + limitOffset + "'";
							throw new SQLNonTransientException(ErrorCode.ER_PARSE_ERROR + " - " + msg);
						} else {
							changedLimit.setOffset(new Int(0));

						}
					}

					query.setLimit(changedLimit);

					String sql = getSql(route, statement, isNeedAddLimit);
					route.updateLimit(sql, 0, limitOffset + limitCount);

					// 设置改写后的sql
					parserResult.setSql(sql);

				} else {

					route.updateLimit(getParserResult().getSql(), route.getLimitStart(), route.getLimitSize());

				}

			}

			route.setCache(isNeedCache(route, schema, query, tableColumnValueSetMapMap));
		}

	}

	/**
	 * 获取所有的条件：因为可能被or语句拆分成多个RouteCalculateUnit，条件分散了
	 * 
	 * @return
	 */
	private Map<String, Map<String, Set<Value>>> getTableColumnValueSetMapMap() {
		Map<String, Map<String, Set<Value>>> tableColumnValueSetMapMap = new HashMap<String, Map<String, Set<Value>>>();
		for (TableColumnValueSetMap tableColumnValueSetMap : parserResult.getTableColumnValueSetMapList()) {
			if (tableColumnValueSetMap != null && tableColumnValueSetMap.getTableColumnValueSetMapMap() != null) {
				tableColumnValueSetMapMap.putAll(tableColumnValueSetMap.getTableColumnValueSetMapMap());
			}
		}

		return tableColumnValueSetMapMap;
	}

	private void tryRoute(Route route, //
			Schema schema, //
			LayeredCache cache) {
		if (route.isRouteOk()) {
			return;// 避免重复路由
		}

		// 无表的select语句直接路由带任一节点
		if ((parserResult.getTableNameList() == null || //
		parserResult.getTableNameList().size() == 0//
		)) {//
			throw new NullPointerException("tableNameList");
		}
		// RouterUtil.tryRouteForTables(schema, ctx, rrs, true, cachePool);
		SortedSet<RouteNode> routeNodeSet = new TreeSet<RouteNode>();
		boolean isAllGlobalTable = RouteUtil.allIsGlobalTable(parserResult, schema);
		for (TableColumnValueSetMap tableColumnValueSetMap : parserResult.getTableColumnValueSetMapList()) {
			// 多个路由信息
			Route tempRoute = RouteUtil.multiTableRoute(route, parserResult, schema, tableColumnValueSetMap, true, cache);
			if (tempRoute != null && tempRoute.getRouteNodes() != null) {
				for (RouteNode routeNode : tempRoute.getRouteNodes()) {
					routeNodeSet.add(routeNode);
				}
			}
			if (isAllGlobalTable) {// 都是全局表时只计算一遍路由
				break;
			}
		}

		if (routeNodeSet.size() == 0) {
			String msg = " find no Route:" + parserResult.getSql();
			LOGGER.error(msg);
			throw new IllegalArgumentException(msg);
		}

		RouteNode[] routeNodes = new RouteNode[routeNodeSet.size()];
		int i = 0;
		for (Iterator<RouteNode> iterator = routeNodeSet.iterator(); iterator.hasNext();) {
			routeNodes[i] = (RouteNode) iterator.next();
			i++;

		}

		route.setRouteNodes(routeNodes);
		route.setFinishedRoute(true);
	}

	protected String getSql(Route route, Statement statement, boolean isNeedAddLimit) {
		if ((isNeedChangeLimit(route) || isNeedAddLimit)) {

			return statement.toString();

		}

		return getParserResult().getSql();
	}

	protected boolean isNeedChangeLimit(Route route) {
		if (route.getRouteNodes() == null) {
			throw new IllegalArgumentException("no node");
		} else {
			if (route.getRouteNodes().length > 1) {
				return true;
			}
			return false;

		}
	}

	private boolean isNeedCache(Route route, Schema schema, Query query, Map<String, Map<String, Set<Value>>> allConditions) {
		if (parserResult.getTableNameList() == null || //
				parserResult.getTableNameList().size() == 0) {//
			return false;
		}
		Table table = schema.getTableMap().get(parserResult.getTableNameList().get(0));
		if (table == null || (parserResult.getTableNameList().size() == 1 && table.isGlobalTable())) {// ||
			return false;
		} else {
			// 单表主键查询
			if (parserResult.getTableNameList().size() == 1) {
				String tableName = parserResult.getTableNameList().get(0);
				String primaryKeyName = schema.getTableMap().get(tableName).getPrimaryKeyName();
				// schema.getTables().get(ctx.getTables().get(0)).getParentKey()
				// != null;
				if (parserResult.getFirstTableColumnValueSetMap().getTableColumnValueSetMapMap().get(tableName) != null && //
						parserResult.getFirstTableColumnValueSetMap().getTableColumnValueSetMapMap().get(tableName).get(primaryKeyName) != null && //
						table.getDataNodeList().size() > 1) {// 有主键条件
					return false;
				}
			}
			return true;
		}
	}

	/**
	 * 单表且是全局表 单表且rule为空且nodeNodes只有一个
	 */
	private boolean isNeedAddLimit(Route route, Schema schema, Query query, Map<String, Map<String, Set<Value>>> tableColumnValueSetMapMap) {
		if (query.getLimit() != null) {// 语句中已有limit
			return false;
		} else if (route.getLimitSize() > -1) {
			return false;
		} else if (schema.getDefaultMaxLimit() == -1) {
			return false;
		} else if (parserResult.getTableNameList().size() == 1) {
			String tableName = parserResult.getTableNameList().get(0);
			Table table = schema.getTableMap().get(tableName);
			if (table == null) {
				// 找不到则取schema的配置
				return schema.getDefaultMaxLimit() > -1;
			}
			boolean isNeedAddLimit = table.isNeedAddLimit();
			if (!isNeedAddLimit) {
				return false;// 优先从配置文件取
			}
			if (schema.getTableMap().get(tableName).isGlobalTable()) {
				return true;
			}
			String primaryKey = schema.getTableMap().get(tableName).getPrimaryKeyName();
			if (tableColumnValueSetMapMap.get(tableName) == null) {// 无条件
				return true;
			}

			if (tableColumnValueSetMapMap.get(tableName).get(primaryKey) != null) {// 条件中带主键
				return false;
			}

			return true;
		} else if (route.hasCachePrimaryKey() && //
				parserResult.getTableNameList().size() == 1) {// 只有一个表且条件中有主键,不需要limit了,因为主键只能查到一条记录
			return false;
		} else {// 多表或无表
			return false;
		}

	}

	private String getAlias(Map<String, String> columnAliasMap, String columnName) {//
		String columnAlias = columnAliasMap.get(columnName);
		if (columnAlias != null) {
			return columnAlias;
		} else {
			if (columnName.indexOf(".") < 0) {
				String columnNameWithDot = "." + columnName;
				// 展开aliaColumns，将<c.name,cname>之类的键值对展开成<c.name,cname>和<name,cname>
				for (Map.Entry<String, String> entry : columnAliasMap.entrySet()) {
					if (entry.getKey().endsWith(columnNameWithDot)) {
						if (entry.getValue() != null && //
								entry.getValue().indexOf(".") > 0) {
							return columnName;
						}
						return entry.getValue();
					}
				}
			}

			return columnName;

		}
	}

	private String[] buildGroupByColumns(//
			List<Expression> groupByItemList,//
			Map<String, String> aliasMap) {//
		String[] groupByColumns = new String[groupByItemList.size()];
		for (int i = 0; i < groupByItemList.size(); i++) {
			Expression groupByItem = groupByItemList.get(i);
			String columnName = null;
			if (groupByItem instanceof Identifier) {
				columnName = ((Identifier) groupByItem).getName();
			} else if (groupByItem instanceof TableAliasName$ColumnName) {
				/**
				 * <pre>
				 * 针对子查询别名，
				 * 例如select id from 
				 * (select h.id from hotnews h 
				 * union
				 * select h.title from hotnews h 
				 * ) as t1 group by t1.id;
				 * </pre>
				 */
				columnName = groupByItem.toString();
			}
			if (columnName == null) {
				columnName = groupByItem.toString();
			}
			int dotIndex = columnName.indexOf(".");
			int bracketIndex = columnName.indexOf("(");
			// 通过判断含有括号来决定是否为函数列
			if (dotIndex != -1 && bracketIndex == -1) {
				// 此步骤得到的column必须是不带.的，
				// 有别名的用别名，无别名的用字段名
				columnName = columnName.substring(dotIndex + 1);
			}
			groupByColumns[i] = getAlias(aliasMap, columnName);// column;
		}
		return groupByColumns;
	}

	protected LinkedHashMap<String, Integer> buildOrderByColumnMap(
	//
			List<OrderByItem> orderByItemList,
			//
			Map<String, String> aliasMap) {
		LinkedHashMap<String, Integer> orderByColumnMap = new LinkedHashMap<String, Integer>();
		for (int i = 0; i < orderByItemList.size(); i++) {
			OrderByType orderByType = orderByItemList.get(i).getOrderByType();
			if (orderByType == null) {
				throw new NullPointerException("orderByType");
			}
			// orderColumn只记录字段名称,因为返回的结果集是不带表名的。
			Expression expression = orderByItemList.get(i).getExpression();
			String columName;
			if (expression instanceof IName) {
				columName = ((IName) expression).getName();
			} else if (expression instanceof TableAliasName$ColumnName) {
				columName = ((TableAliasName$ColumnName) expression).getName();
			} else {
				throw new NullPointerException("columName");
			}

			columName = getAlias(aliasMap, columName);// 此步骤得到的col必须是不带.的，有别名的用别名，无别名的用字段名
			orderByColumnMap.put(columName, //
					orderByType == OrderByType.ASC ? //
					OrderByColum.ORDER_TYPE_ASC
							: //
							OrderByColum.ORDER_TYPE_DESC);//
		}
		return orderByColumnMap;
	}

	@SuppressWarnings("unused")
	private boolean isConditionAlwaysTrue(Statement statement) {
		SelectStatement selectStatement = (SelectStatement) statement;
		IQuery iQuery = selectStatement.getSubQuery().getIQuery();
		if (iQuery instanceof Query) {
			Query query = (Query) selectStatement.getSubQuery().getIQuery();
			Expression where = query.getWhere();

			Object object = VisitorUtils.getValue(where);
			if (Boolean.TRUE.equals(object)) {
				return true;
			}
			return false;
		} else {// union
			return false;
		}

	}

}
